package com.shujia.streaming

import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark.streaming.{Durations, StreamingContext}
import org.apache.spark.streaming.dstream.DStream

object Demo04WindowOpt {
  def main(args: Array[String]): Unit = {
    /**
     * 微博热搜：每10分钟基于最近24小时的热度进行排名
     *
     * 简化：
     * 每隔5s钟统计最近10s内的单词数量 --> 滑动窗口
     * 每隔5s钟统计最近5s内的单词数量 --> 滚动窗口
     *
     * 两个时间：滑动时间、窗口大小
     * 滑动时间 = 窗口大小 即为 滚动窗口
     * 滑动时间 < 窗口大小 即为 滑动窗口
     * 滑动时间 > 窗口大小 一般不用 适用场景极少
     */
    val conf: SparkConf = new SparkConf()
    conf.setAppName("Demo04WindowOpt")
    conf.setMaster("local[2]")
    val sc: SparkContext = new SparkContext(conf)
    // 每5s一个批次 即 滑动时间
    val ssc: StreamingContext = new StreamingContext(sc, Durations.seconds(5))

    // 如果使用了有状态算子或者用到了之前的结果则需要设置checkPoint的目录
    ssc.checkpoint("spark/data/ssc/ck")

    val lineDS: DStream[String] = ssc.socketTextStream("master", 8888)

    lineDS
      .flatMap(_.split(","))
      .map(word => (word, 1))
      /**
       * 优化：
       * 例如每5s统计最近10s内的单词数量
       *
       * 第10秒时 统计的是 0~5s + 5s~10s 内的单词数量
       * 第15秒时 统计的是 5s~10s + 10s~15s 内的单词数量
       *
       * 所以在计算第15秒时的结果时可以用第10秒计算出来的结果 减去0~5s的结果 再 加上10s~15s内的结果
       */
      .reduceByKeyAndWindow(_ + _, _ - _, Durations.seconds(10)) // 指定窗口大小
      .print()

    ssc.start()
    ssc.awaitTermination()
    ssc.stop()


  }

}
